package openhaven.control.net;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.ListIterator;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;

import openhaven.control.message.RichMessage;

import org.apache.log4j.Logger;

public class Session
{
    private static final Logger log = Logger.getLogger(Session.class);
    public static final int PVER = 25;

    public static final int MSG_SESS = 0;
    public static final int MSG_REL = 1;
    public static final int MSG_ACK = 2;
    public static final int MSG_BEAT = 3;
    public static final int MSG_MAPREQ = 4;
    public static final int MSG_MAPDATA = 5;
    public static final int MSG_OBJDATA = 6;
    public static final int MSG_OBJACK = 7;
    public static final int MSG_CLOSE = 8;
    public static final int OD_REM = 0;
    public static final int OD_MOVE = 1;
    public static final int OD_RES = 2;
    public static final int OD_LINBEG = 3;
    public static final int OD_LINSTEP = 4;
    public static final int OD_SPEECH = 5;
    public static final int OD_LAYERS = 6;
    public static final int OD_DRAWOFF = 7;
    public static final int OD_LUMIN = 8;
    public static final int OD_AVATAR = 9;
    public static final int OD_FOLLOW = 10;
    public static final int OD_HOMING = 11;
    public static final int OD_OVERLAY = 12;
    public static final int OD_AUTH = 13;
    public static final int OD_HEALTH = 14;
    public static final int OD_END = 255;
    public static final int SESSERR_AUTH = 1;
    public static final int SESSERR_BUSY = 2;
    public static final int SESSERR_CONN = 3;
    public static final int SESSERR_PVER = 4;
    public static final int SESSERR_EXPR = 5;

    static final int ackthresh = 30;

    private DatagramSocket sk;
    private final InetAddress server;
    private final Thread rworker, sworker, ticker;
    public int connfailed = 0;
    public String state = "conn";
    private int tseq = 0;

    private int rseq = 0;
    private int ackseq;
    private long acktime = -1;

    // This is a queue holding messages that need to be sent to the UI
    private final BlockingQueue<Message> uimsgs;

    // waiting is a map to collect messages that we have received ahead of time
    private final Map<Integer, Message> waiting = new TreeMap<Integer, Message>();

    private final LinkedList<Message> pending = new LinkedList<Message>();
    private final Map<Integer, ObjAck> objacks = new TreeMap<Integer, ObjAck>();
    private final String username;
    private final byte[] cookie;

    @SuppressWarnings("serial")
    public class MessageException extends RuntimeException
    {
        public Message msg;

        public MessageException(final String text, final Message msg)
        {
            super(text);
            this.msg = msg;
        }
    }

    private class ObjAck
    {
        int id;
        int frame;
        long recv;
        long sent;

        public ObjAck(final int id, final int frame, final long recv)
        {
            this.id = id;
            this.frame = frame;
            this.recv = recv;
            sent = 0;
        }
    }

    private class Ticker extends Thread
    {
        public Ticker()
        {
            super(tg(), "Server time ticker");
            setDaemon(true);
        }

        @Override
        public void run()
        {
            try
            {
                while (true)
                {
                    long now, then;
                    then = System.currentTimeMillis();
                    // glob.oc.tick();
                    now = System.currentTimeMillis();
                    if (now - then < 70)
                    {
                        Thread.sleep(70 - (now - then));
                    }
                }
            }
            catch (final InterruptedException e)
            {
            }
        }
    }

    private class RWorker extends Thread
    {
        boolean alive;

        public RWorker()
        {
            super(tg(), "Session reader");
            setDaemon(true);
        }

        private void gotack(final int seq)
        {
            synchronized (pending)
            {
                for (final ListIterator<Message> i = pending.listIterator(); i.hasNext();)
                {
                    final Message msg = i.next();
                    if (msg.seq <= seq)
                    {
                        i.remove();
                    }
                }
            }
        }

        private void getobjdata(final Message msg)
        {
            // used to use the global object cache thinger to update objects in
            // the model directly
            while (msg.off < msg.blob.length)
            {
                final int id = msg.int32();
                final int frame = msg.int32();
                synchronized (objacks)
                {
                    if (objacks.containsKey(id))
                    {
                        final ObjAck a = objacks.get(id);
                        a.frame = frame;
                        a.recv = System.currentTimeMillis();
                    }
                    else
                    {
                        objacks.put(id, new ObjAck(id, frame, System.currentTimeMillis()));
                    }
                }
                synchronized (sworker)
                {
                    sworker.notifyAll();
                }
            }
        }

        private void handlerel(final Message msg)
        {
            uimsgs.add(msg);
        }

        private void getrel(final int seq, final Message msg)
        {
            if (seq == rseq)
            {
                synchronized (uimsgs)
                {
                    // this is the right sequence, handle the message
                    handlerel(msg);

                    // handle everything else we've already received
                    while (true)
                    {
                        // increment the sequence
                        rseq = (rseq + 1) % 65536;

                        // haven't received the next message yet
                        if (!waiting.containsKey(rseq))
                        {
                            break;
                        }
                        // already have the next message, handle it then remove
                        // from the map
                        handlerel(waiting.get(rseq));
                        waiting.remove(rseq);
                    }
                }
                // tell the server what we've received up to, it should send us
                // back a message with sequence rseq
                sendack(rseq - 1);

                // make sure the send worker is alive (lazy bastard keeps going
                // to sleep when there's nothing to send... THE NERVE!)
                synchronized (Session.this)
                {
                    Session.this.notifyAll();
                }
            }
            else if (seq > rseq)
            {
                // message received out of order, put it in the waiting map and
                // we'll handle it when it's sequence comes up
                waiting.put(seq, msg);
            }
        }

        @Override
        public void run()
        {
            try
            {
                alive = true;
                try
                {
                    sk.setSoTimeout(1000);
                }
                catch (final SocketException e)
                {
                    throw (new RuntimeException(e));
                }
                while (alive)
                {
                    final DatagramPacket p = new DatagramPacket(new byte[65536], 65536);
                    try
                    {
                        sk.receive(p);
                    }
                    catch (final java.nio.channels.ClosedByInterruptException e)
                    {
                        /*
                         * Except apparently Sun's J2SE doesn't throw this when
                         * interrupted :P
                         */
                        break;
                    }
                    catch (final SocketTimeoutException e)
                    {
                        continue;
                    }
                    catch (final IOException e)
                    {
                        throw (new RuntimeException(e));
                    }
                    if (!p.getAddress().equals(server))
                    {
                        continue;
                    }
                    final Message msg = new Message(p.getData()[0], p.getData(), 1, p.getLength() - 1);
                    log.debug("RWorker got message: ");
                    if (msg.type == MSG_SESS)
                    {
                        log.debug("session message");
                        if (state == "conn")
                        {
                            final int error = msg.uint8();
                            synchronized (Session.this)
                            {
                                if (error == 0)
                                {
                                    state = "";
                                }
                                else
                                {
                                    connfailed = error;
                                    close();
                                }
                                Session.this.notifyAll();
                            }
                        }
                    }
                    if (state != "conn")
                    {
                        if (msg.type == MSG_SESS)
                        {
                            log.debug("session message state != con");
                        }
                        else if (msg.type == MSG_REL)
                        {
                            int seq = msg.uint16();
                            while (!msg.eom())
                            {
                                log.debug("rel message");
                                int type = msg.uint8();
                                int len;
                                if ((type & 0x80) != 0)
                                {
                                    type &= 0x7f;
                                    len = msg.uint16();
                                }
                                else
                                {
                                    len = msg.blob.length - msg.off;
                                }
                                getrel(seq, new Message(type, msg.blob, msg.off, len));
                                msg.off += len;
                                seq++;
                            }
                        }
                        else if (msg.type == MSG_ACK)
                        {
                            log.debug("ACK!");
                            gotack(msg.uint16());
                        }
                        else if (msg.type == MSG_MAPDATA)
                        {
                            log.debug("map data");
                            // glob.map.mapdata(msg);
                        }
                        else if (msg.type == MSG_OBJDATA)
                        {
                            log.debug("object data");
                            getobjdata(msg);
                        }
                        else if (msg.type == MSG_CLOSE)
                        {
                            log.debug("close message");
                            synchronized (Session.this)
                            {
                                state = "fin";
                            }
                            close();
                        }
                        else
                        {
                            throw (new MessageException("Unknown message type: " + msg.type, msg));
                        }
                    }
                }
            }
            finally
            {
                synchronized (Session.this)
                {
                    log.debug("session dead");
                    state = "dead";
                    Session.this.notifyAll();
                }
            }
        }

        @Override
        public void interrupt()
        {
            alive = false;
            super.interrupt();
        }
    }

    private class SWorker extends Thread
    {

        public SWorker()
        {
            super(tg(), "Session writer");
            setDaemon(true);
        }

        @Override
        public void run()
        {
            try
            {
                long to, last = 0, retries = 0;
                while (true)
                {
                    long now = System.currentTimeMillis();
                    if (state == "conn")
                    {
                        if (now - last > 2000)
                        {
                            if (++retries > 5)
                            {
                                synchronized (Session.this)
                                {
                                    connfailed = SESSERR_CONN;
                                    Session.this.notifyAll();
                                    return;
                                }
                            }
                            final Message msg = new Message(MSG_SESS);
                            msg.adduint16(PVER);
                            msg.addstring(username);
                            msg.addbytes(cookie);
                            sendmsg(msg);
                            last = now;
                        }
                        Thread.sleep(100);
                    }
                    else
                    {
                        to = 5000;
                        synchronized (pending)
                        {
                            if (pending.size() > 0)
                            {
                                to = 60;
                            }
                        }
                        synchronized (objacks)
                        {
                            if ((objacks.size() > 0) && (to > 120))
                            {
                                to = 200;
                            }
                        }
                        synchronized (this)
                        {
                            if (acktime > 0)
                            {
                                to = acktime + ackthresh - now;
                            }
                            if (to < 0)
                            {
                                to = 0;
                            }
                            this.wait(to);
                        }
                        now = System.currentTimeMillis();
                        boolean beat = true;
                        /*
                         * if((closing != -1) && (now - closing > 500)) {
                         * Message cm = new Message(MSG_CLOSE); sendmsg(cm);
                         * closing = now; if(++ctries > 5)
                         * getThreadGroup().interrupt(); }
                         */
                        synchronized (pending)
                        {
                            if (pending.size() > 0)
                            {
                                for (final Message msg : pending)
                                {
                                    if (now - msg.last > 60)
                                    { /* XXX */
                                        msg.last = now;
                                        final Message rmsg = new Message(MSG_REL);
                                        rmsg.adduint16(msg.seq);
                                        rmsg.adduint8(msg.type);
                                        rmsg.addbytes(msg.blob);
                                        sendmsg(rmsg);
                                    }
                                }
                                beat = false;
                            }
                        }
                        synchronized (objacks)
                        {
                            Message msg = null;
                            for (final Iterator<ObjAck> i = objacks.values().iterator(); i.hasNext();)
                            {
                                final ObjAck a = i.next();
                                boolean send = false, del = false;
                                if (now - a.sent > 200)
                                {
                                    send = true;
                                }
                                if (now - a.recv > 120)
                                {
                                    send = del = true;
                                }
                                if (send)
                                {
                                    if (msg == null)
                                    {
                                        msg = new Message(MSG_OBJACK);
                                    }
                                    msg.addint32(a.id);
                                    msg.addint32(a.frame);
                                    a.sent = now;
                                }
                                if (del)
                                {
                                    i.remove();
                                }
                            }
                            if (msg != null)
                            {
                                sendmsg(msg);
                                beat = false;
                            }
                        }
                        synchronized (this)
                        {
                            if ((acktime > 0) && (now - acktime >= ackthresh))
                            {
                                final byte[] msg = { MSG_ACK, 0, 0 };
                                BitFiddlingUtils.uint16e(ackseq, msg, 1);
                                sendmsg(msg);
                                acktime = -1;
                                beat = false;
                            }
                        }
                        if (beat)
                        {
                            if (now - last > 5000)
                            {
                                sendmsg(new byte[] { MSG_BEAT });
                                last = now;
                            }
                        }
                    }
                }
            }
            catch (final InterruptedException e)
            {
                for (int i = 0; i < 5; i++)
                {
                    synchronized (Session.this)
                    {
                        if ((state == "conn") || (state == "fin"))
                        {
                            break;
                        }
                    }
                    sendmsg(new Message(MSG_CLOSE));
                    final long f = System.currentTimeMillis();
                    while (true)
                    {
                        final long now = System.currentTimeMillis();
                        if (now - f > 500)
                        {
                            break;
                        }
                        try
                        {
                            Thread.sleep(500 - (now - f));
                        }
                        catch (final InterruptedException e2)
                        {
                        }
                    }
                }
            }
            finally
            {
                ticker.interrupt();
                rworker.interrupt();
            }
        }
    }

    public Session(final InetAddress server, final String username, final byte[] cookie, final BlockingQueue<Message> uiMessageQueue)
    {
        this.server = server;
        this.username = username;
        this.cookie = cookie;
        uimsgs = uiMessageQueue;
        try
        {
            sk = new DatagramSocket();
        }
        catch (final SocketException e)
        {
            throw (new RuntimeException(e));
        }
        rworker = new RWorker();
        rworker.start();
        sworker = new SWorker();
        sworker.start();
        ticker = new Ticker();
        ticker.start();
    }

    private void sendack(final int seq)
    {
        synchronized (sworker)
        {
            if (acktime < 0)
            {
                acktime = System.currentTimeMillis();
            }
            ackseq = seq;
            sworker.notifyAll();
        }
    }

    public void close()
    {
        sworker.interrupt();
    }

    public synchronized boolean alive()
    {
        return (state != "dead");
    }

    public void queuemsg(final Message msg)
    {
        log.debug("queueing up message" + RichMessage.process(msg));
        msg.seq = tseq;
        tseq = (tseq + 1) % 65536;
        synchronized (pending)
        {
            pending.add(msg);
        }
        synchronized (sworker)
        {
            sworker.notify();
        }
    }

    public void sendmsg(final Message msg)
    {
        log.debug("sending message type:" + Message.messageTypeBinding.get(msg.type) + RichMessage.process(msg));
        final byte[] buf = new byte[msg.blob.length + 1];
        buf[0] = (byte) msg.type;
        System.arraycopy(msg.blob, 0, buf, 1, msg.blob.length);
        sendmsg(buf);
    }

    public void sendmsg(final byte[] msg)
    {
        try
        {
            sk.send(new DatagramPacket(msg, msg.length, server, 1870));
        }
        catch (final IOException e)
        {
            e.printStackTrace();
        }
    }

    public static ThreadGroup tg()
    {
        return (Thread.currentThread().getThreadGroup());
    }
}
